package xyz.hubery.handler;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Recover;
import org.springframework.retry.annotation.Retryable;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import xyz.hubery.domain.AsyncTask;
import xyz.hubery.enums.AsyncTaskStatus;
import xyz.hubery.enums.AsyncTaskType;
import xyz.hubery.mapper.AsyncTaskMapper;
import xyz.hubery.model.Constant;
import xyz.hubery.model.TaskResult;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 异步任务调度器（控制数据流转流程，自动选择合适的任务处理器处理任务
 */
@Component
@Slf4j
@ConditionalOnProperty(name = "async.task.scheduler.enabled", havingValue = "true")
public class AsyncTaskScheduler {
    
    @Autowired
    private List<AsyncTaskHandler> handlers;
    
    @Autowired
    private AsyncTaskMapper asyncTaskMapper;
    
    private final Map<AsyncTaskType, AsyncTaskHandler> handlerMap = new HashMap<>();

    // 完成依赖注入后执行
    @PostConstruct
    public void init() {
        // 初始化处理器映射（这样我们就可以根据任务类型获取到对应的任务处理器
        if(CollectionUtils.isEmpty(handlers)) {
            return;
        }
        for (AsyncTaskHandler handler : handlers) {
            handlerMap.put(handler.getSupportedType(), handler);
        }
    }


    // 自定义线程池
    @Bean("taskExecutor")
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);     // 核心线程数
        executor.setMaxPoolSize(10);     // 最大线程数
        executor.setQueueCapacity(50);   // 队列容量
        executor.setThreadNamePrefix("async-task-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }

    @Autowired
    @Qualifier("taskExecutor")
    private ThreadPoolTaskExecutor taskExecutor;
    
    /**
     * 处理待处理的任务
     */
    @Scheduled(fixedDelay = 5000) // 每5秒执行一次
    public void processPendingTasks() {
        log.info("定时任务开始...");
        //  获取待处理任务
        //  TODO 后续可改进分批获取任务，避免一次性处理太多
        List<AsyncTask> tasks = asyncTaskMapper.findPendingTasks();

        // 控制并发数量，避免数据库连接耗尽
        int batchSize = Math.min(tasks.size(), taskExecutor.getMaxPoolSize() * 2);
        List<AsyncTask> batchTasks = tasks.subList(0, batchSize);

        // 让当前线程等待子线程完成后在执行
        CountDownLatch latch = new CountDownLatch(batchTasks.size());

        for (AsyncTask task : batchTasks) {

            taskExecutor.execute(()->{
                try {

                    processSingleTask(task); // 处理单个任务
                } catch (Exception e) {
                    log.error("处理任务异常, taskId: {}", task.getId(), e);
                }finally {
                    latch.countDown(); // 每执行完一个任务-1
                }
            });
        }

        try {
            // 等待这批任务完成，但不超过30秒
            latch.await(30, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.warn("任务处理被中断");
        }

        log.info("本批次任务处理完成");
    }
    
    /**
     * 处理单个任务
     */
    @Retryable(
            value = {Exception.class},
            maxAttemptsExpression = "#{@asyncTaskManager.getMaxRetry(#root.args[0])}",
            backoff = @Backoff(delay = 2000, multiplier = 2)
    )
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void processSingleTask(AsyncTask task) {
        // 选择合适的处理器
        AsyncTaskHandler handler = handlerMap.get(AsyncTaskType.of(task.getType()));
        if (handler == null) { // 这种不需要重试
            log.warn("未找到对应的任务处理器, type: {}", task.getType());
            handler.onFailure(task,TaskResult.failed("未找到对应的任务处理器:"+task.getType()));
            return;
        }
        
        // 检查重试次数
        if(task.getRetryCount() >= handler.getMaxRetry(task)) {
            log.warn("重试次数超出限制, 任务id: {}", task.getId());
            handler.onFailure(task,TaskResult.failed("重试次数过多"));
            return;
        }
        // 检查任务状态
        if(AsyncTaskStatus.PENDING.getCode() != task.getStatus()) { // 当前任务不需要处理（被其他线程处理了
            log.warn("当前任务不需要处理, 任务id: {}", task.getId());
            handler.onFailure(task,TaskResult.failed("已经被处理过:"+task.getId()));
        }

        try {
            task.setStatus(AsyncTaskStatus.PROCESSING.getCode()); //设置状态为处理中，避免其他线程并发执行
            TaskResult result = handler.handle(task); // 执行具体的业务逻辑，交由专门的任务处理器处理
            
            switch (result.getStatus()) {
                case SUCCESS: // 处理成功
                    handler.onSuccess(task,result);
                    break;
                case RETRY: // 需要重试
                    handler.onRetry(task);
                    break;
                case FAILED: // 处理失败
                    handler.onFailure(task,result);
                    break;
                case SKIPPED: // 跳过
                    handler.onSkipped(task, result.getMessage());
                    break;
            }

        } catch (Exception e) {
            log.error("任务处理异常, taskId: {}", task.getId(), e); //重试
            handler.onRetry(task);
        }
    }

    /**
     * 重试耗尽后的恢复方法
     */
    @Recover
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void recoverProcessSingleTask(Exception e, AsyncTask task) {
        log.error("任务处理重试耗尽, taskId: {}", task.getId(), e);

        AsyncTaskHandler handler = handlerMap.get(AsyncTaskType.of(task.getType()));
        if (handler != null) {
            // 调用最终失败回调
            handler.onFailure(task, TaskResult.failed("重试耗尽: " + e.getMessage()));
        }
    }

//    /**
//     * 发生了重试
//     * @param task
//     * @param handler
//     */
//    private void scheduleRetry(AsyncTask task, AsyncTaskHandler handler) {
//        task.setRetryCount(task.getRetryCount() + 1);
//        task.setStatus(AsyncTaskStatus.PENDING.getCode()); // 回退至待处理状态
//        asyncTaskMapper.updateById(task);
//    }
//
//    /**
//     * 计算下一次重试时间
//     * @param task
//     * @param handler
//     * @return
//     */
//    private Date calculateNextRetryTime(AsyncTask task, AsyncTaskHandler handler) {
//        // 指数退避策略
//        long delay = handler.getRetryInterval() * (long) Math.pow(2, task.getRetryCount() - 1);
//        return new Date(System.currentTimeMillis() + delay);
//    }
//
//    /**
//     * 任务处理成功
//     * @param task
//     */
//    private void markTaskSuccess(AsyncTask task) {
//        task.setStatus(AsyncTaskStatus.SUCCESS.getCode());
//        asyncTaskMapper.updateById(task);
//    }
//
//    /**
//     * 任务处理失败时执行
//     * @param task
//     * @param reason
//     */
//    private void markTaskFailed(AsyncTask task, String reason) {
//        task.setStatus(AsyncTaskStatus.FAILED.getCode());
//        task.setMsg(reason);
//        asyncTaskMapper.updateById(task);
//        // 发送告警
//    }
}